package com.stillcoolme.framework.hadoop;


import com.stillcoolme.framework.util.Mean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;

public class GetMaxValue {

    public static class MapClass extends Mapper<LongWritable, Text, Text, DoubleWritable> {
		public void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
			String[] str = value.toString().split(",");

            Double score = Double.parseDouble(str[8]);

            context.write(new Text("Max"), new DoubleWritable(score));
		}
	}

	public static class Reduce extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
		private double maxNum = 0;
//        private double minNum = 101;
		private Text one = new Text();
		public void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
            System.out.println(values);
			for (DoubleWritable val : values) {
				if ( val.get() > maxNum) {
					maxNum = val.get();
				}
			}
            System.out.println("$$$");
            System.out.println(maxNum);
			one = key;
		}
		@Override
		protected void cleanup(Context context) throws IOException,
		InterruptedException {
			context.write(one, new DoubleWritable(maxNum));
		}
	}
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration(true);
        configuration.set("fs.defaultFS","hdfs://hadoop102:8020");
        //本地模式运行
        configuration.set("mapreduce.framework.name", "local");
        //创建Job
        Job job = Job.getInstance(configuration);
        job.setJobName("Max_min"+ new SimpleDateFormat("yyyyMMdd-HHmmss-SSS").format(new Date()));
        job.setJarByClass(Mean.class);


        job.setNumReduceTasks(1);

        FileInputFormat.setInputPaths(job, new Path("/score/data/2020jsj.csv"));
        FileOutputFormat.setOutputPath(job, new Path("/score/result/" + job.getJobName()));


        job.setMapperClass(MapClass.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);

        job.setReducerClass(Reduce.class);


        job.waitForCompletion(true);

    }

}
